/** @file $RCSfile: tbusd_send.c,v $
  general description of this module
  $Id: tbusd_send.c,v 1.1.1.1 2013/05/27 02:58:11 ouyang Exp $
@author $Author: ouyang $
@date $Date: 2013/05/27 02:58:11 $
@version $Revision: 1.1.1.1 $
@note Editor: Vim 6.3, Gcc 4.0.2, tab=4
@note Platform: Linux
*/



#include "tbusd_log.h"
#include "tbusd_send.h"
#include "tbusd_misc.h"
#include "tbus/tbus_dyemsg.h"
#include "tbus/tbus_channel.h"
#include "tbus/tbus_log.h"


extern RELAYENV gs_stRelayEnv;

/** 在relay通道上发送数据
@param[in] a_pstRunEnv -- 服务器运行上下文环境
@param[in] a_pstRelay relay通道数据结构指针
@param[out] a_iMaxSendPkg 最大可以发送数据包

@retval 0 -- no package send
@retval >0 number of package sended  

@note
*/
int tbusd_send_pkg(IN LPRELAYENV a_pstRunEnv, IN LPRELAY a_pstRelay, IN int a_iMaxSendPkg)
{
	int iRet = TBUS_SUCCESS;
	int	iPkgSended = 0;
	LPCONNNODE pstConn;
	LPTBUSCHANNEL  pstChannel;
	LPTBUSRELAYCONNSTAT pstRelayStat;
	int iHeadSeq = 0;
	int iGetSeq = 0;
	int iAckNum ;
	char *sData = NULL ;
	CHANNELVAR *pstVar = NULL ;
	LPTBUSD pstConf ;
	


	assert(NULL != a_pstRunEnv);
	assert(NULL != a_pstRelay);
	assert(NULL != a_pstRelay->pConnNode);
	assert(NULL != a_pstRunEnv->pstStat);
	

	pstConf = a_pstRunEnv->pstConf;
	assert(NULL != pstConf);
	pstConn = a_pstRelay->pConnNode;
	pstChannel = &a_pstRelay->stChl;
	assert(a_pstRelay->iID < (int)a_pstRunEnv->dwRelayCnt);
	pstRelayStat = &a_pstRunEnv->pstStat->astRelayConnStat[a_pstRelay->iID];

	/*先发送剩余没有发送完的数据*/
	iRet = tbusd_send_remain_data(pstConn, pstChannel, pstRelayStat) ;
	if ( 0 != iRet )
	{
		tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_ERROR,
			"tbusd_send_pkg, tbusd_send_remain_data failed to send remain data on socket %d of relay %d, ret %d",
			pstConn->iFd, a_pstRelay->iID, iRet ) ;
		return 0;
	}
	if (TBUSD_CONN_HAVE_REMAIN_DATA(pstConn))
	{
		return 0;	/*同步数据还没有发送完,则先不发送业务数据*/
	}

	/*发送seq同步数据*/
	if (0 != pstConf->iNeedAckPkg)
	{
		iRet = tbusd_send_synseq_pkg(pstConn, a_pstRelay, pstRelayStat);
		if ( 0 != iRet )
		{
			tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_ERROR,
				"tbusd_send_pkg, tbusd_send_synseq_pkg failed to send sysseq pkg on socket %d of relay %d, ret %d",
				pstConn->iFd, a_pstRelay->iID, iRet ) ;
			return 0;
		}
		if (TBUSD_CONN_HAVE_REMAIN_DATA(pstConn))
		{
			return 0;	/*同步数据还没有发送完,则先不发送业务数据*/
		}
	}/*if (0 != pstConf->iNeedAckPkg)*/
	
	
	/*设计更改，发送的最大包数目由可发送最大为确认报数计算出一个最大值限制,这样对未确认包作
	进行流量控制*/
	a_iMaxSendPkg = (a_pstRunEnv->iMaxSendPkgPerLoop < a_iMaxSendPkg)? a_iMaxSendPkg : a_pstRunEnv->iMaxSendPkgPerLoop;
	if ((0 != pstConf->iNeedAckPkg) && (pstConf->iControlUnackedPkgNum != 0))
	{
		pstVar = TBUS_CHANNEL_VAR_GET(pstChannel);
		sData = TBUS_CHANNEL_QUEUE_GET(pstChannel);
		iHeadSeq = tbus_get_pkgseq_by_pos(pstChannel, pstVar->dwHead);
		iGetSeq = tbus_get_pkgseq_by_pos(pstChannel, pstVar->dwGet);
		iAckNum = iGetSeq - iHeadSeq;/*未确认数目包数目*/
		if (iAckNum >= TBUS_MAX_SENDED_PKG_NUM_UNACK)
		{
			tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_ERROR,
				"unacked Sended pkg number(%d) reach the max limit(%d), so stop to send pkg!!!"
				"unacked pkg seq form %d to %d",
				iAckNum, TBUS_MAX_SENDED_PKG_NUM_UNACK, iHeadSeq, iGetSeq);
			return 0;
		}
		iAckNum = TBUS_MAX_SENDED_PKG_NUM_UNACK - iAckNum; /*计算可发送未确认数据包数目*/
		a_iMaxSendPkg = (a_iMaxSendPkg > iAckNum) ? iAckNum : a_iMaxSendPkg;
	}/*if (pstConf->iControlUnackedPkgNum != 0)*/
	
	
	

	/*发送业务数据*/
	while(iPkgSended < a_iMaxSendPkg)
	{
		/*取出一个消息,发送发送缓冲区中*/
		iRet = tbusd_send_one_pkg(pstChannel, pstConn, pstRelayStat);
		if (0 != iRet)
		{
			break;
		}

		if (TBUSD_CONN_HAVE_REMAIN_DATA(pstConn))
		{
			break; /*当前数据包还没有发送完毕,则不发下一个数据*/
		}

		iPkgSended++;		
	}/*while(iPkgSended < a_iMaxSendPkg)*/


	if (iPkgSended >  (int)pstRelayStat->dwMaxSendCountPerloop)
	{
		pstRelayStat->dwMaxSendCountPerloop = iPkgSended;
	}
	return iPkgSended;
}


int tbusd_send_remain_data(INOUT LPCONNNODE a_pstConn,INOUT LPTBUSCHANNEL a_pstChannel, LPTBUSRELAYCONNSTAT a_pstRelayStat)
{

	int	iRemain = 0;
	char *pszDataToSend;
	int	iSend = 0 ;
	LPSENDDATAMNG pstSendMng ;

	assert(NULL != a_pstConn);
	assert(NULL != a_pstChannel);

	if (!TBUSD_CONN_HAVE_REMAIN_DATA(a_pstConn))
	{
		return TBUS_SUCCESS;	/*同步数据还没有发送完,则先不发送业务数据*/
	}
		
	pstSendMng = &a_pstConn->stSendMng;
	tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_TRACE, 
		"tbusd_send_remain_data, pkg<len:%d> seq: %u type: %d sendPos: %d on socket %d",
		pstSendMng->iSendBuff, pstSendMng->stSendHead.iSeq, pstSendMng->enRemainDataType, 
		pstSendMng->iSendPos, a_pstConn->iFd);
	

	/*发送数据*/
	if (pstSendMng->enRemainDataType == TBUSD_DATA_TYPE_SYN)
	{
		pszDataToSend = pstSendMng->pszSynData + pstSendMng->iSendPos;
	}else
	{
		pszDataToSend = pstSendMng->pszBussiData + pstSendMng->iSendPos;
	}
	iRemain = pstSendMng->iSendBuff - pstSendMng->iSendPos;
	iSend = tbusd_send(a_pstConn->iFd, pszDataToSend, iRemain);
	if( iSend<0 )
	{
		tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_ERROR, 
			"tbusd_send_remain_data, send remain data failed on socket %d, so close socket",
			a_pstConn->iFd);

		a_pstConn->bNeedFree = TBUSD_TRUE;
		return -1 ;
	}

	/*检查发送结果*/
	a_pstRelayStat->ullSendPkgBytes += iSend;
	if( iSend < iRemain )
	{	/*还有数据没有发送完成*/
		pstSendMng->iSendPos += iSend;
		return TBUS_SUCCESS;
	}

	/*剩余数据已经发送完,则修改相应数据结构*/
	switch(pstSendMng->enRemainDataType)
	{
	case TBUSD_DATA_TYPE_SYN:
		time(&a_pstConn->tLastSynSeq); /*更新最后一次发送seq同步包的时间*/
		a_pstRelayStat->dwSendPkgCount++;
		break;
	case TBUSD_DATA_TYPE_BUSINESS:
		if(pstSendMng->stSendHead.bFlag & TBUS_HEAD_FLAG_TACE)
		{
			tbus_log_dyedpkg(&pstSendMng->stSendHead, "Send(tbusd)");
		}
		tbusd_adjust_sending_channel_pos(pstSendMng,a_pstChannel);
		a_pstRelayStat->dwSendPkgCount++;
		break;
	default:
		break;
	}
	pstSendMng->iSendPos = 0;
	pstSendMng->iSendBuff = 0;
	pstSendMng->enRemainDataType = TBUSD_DATA_TYPE_UNKNOWN;
	
	return TBUS_SUCCESS;
}



int tbusd_adjust_sending_channel_pos(LPSENDDATAMNG a_pstSendMng,  IN const TBUSCHANNEL *a_pstChannel) 
{
	CHANNELVAR *pstVar = NULL ;

	assert(NULL != a_pstChannel);
	assert(NULL != a_pstSendMng);

	pstVar = TBUS_CHANNEL_VAR_GET(a_pstChannel);
	
	/*检查是否已经移到队列末尾*/
	if (pstVar->dwGet == pstVar->dwTail)
	{
		return TBUS_SUCCESS;
	}
	
	/*修改移动指针*/
	tbus_moveto_next_pkg(&a_pstSendMng->stSendHead, a_pstChannel->pstHead, pstVar->dwSize, &pstVar->dwGet, pstVar->dwTail);
	
	if (!pstVar->chGStart)
	{
		pstVar->chGStart = 1;
	}

	/*设计更改：缺省tbusd通过对发送出去数据包序列号确认提供可靠数据传输，但应部分业务需求希望tbusd提供提供容许丢包的
	的版本，因此tbusd的配置中增加一个配置，通过这个配置控制是否需要对发送数据包作确认，如果不需要确认则除移动get指针外
	直接通过序列号移动head指针*/
	assert(NULL != gs_stRelayEnv.pstConf);
	if (0 == gs_stRelayEnv.pstConf->iNeedAckPkg)
	{
		/*不需要对发送的数据包做确认，直接调用tbusd_syn_seq确认数据包，以移动head指针*/
		tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_TRACE, "Tbusd need not ack sended pkg,"
			" so call tbusd_syn_seq ack seq: %u", a_pstSendMng->stSendHead.iSeq);
		tbusd_syn_seq(a_pstChannel, a_pstSendMng->stSendHead.iSeq);
	}

	return 0;
}


int tbusd_send_one_pkg( IN const TBUSCHANNEL *a_pstChannel, IN LPCONNNODE a_pstConn, 
					   LPTBUSRELAYCONNSTAT a_pstRelayStat)
{
	int iPkgLen = 0,
		iDataLen = 0;
	char *sData = NULL ;
	CHANNELVAR *pstVar = NULL ;
	CHANNELVAR *pstPushVar = NULL ;
	int iRet = 0;
	LPTBUSHEAD pstHead;


	pstVar = TBUS_CHANNEL_VAR_GET(a_pstChannel);
	sData = TBUS_CHANNEL_QUEUE_GET(a_pstChannel);
	if (pstVar->dwGet == pstVar->dwTail)
	{
		return TBUS_ERR_CHANNEL_EMPTY;
	}


	/*检查从头指针到队列末尾是否足够保存一条消息*/
	TBUS_CHECK_QUEUE_HEAD_VAR(sData, pstVar->dwSize, pstVar->dwGet);
	iDataLen = (int)(pstVar->dwTail - pstVar->dwGet);
	if (iDataLen < 0) 
	{
		iDataLen += pstVar->dwSize;
	}
	if (0 >= iDataLen) 
	{   	/*再次检查数据队列中是否有数据*/
		return TBUS_ERR_CHANNEL_EMPTY;
	}

	/*get head */
	pstHead = &a_pstConn->stSendMng.stSendHead;
	iRet = tbus_decode_head(pstHead, sData +pstVar->dwGet, iDataLen, 0);
	if (TBUS_SUCCESS != iRet)
	{
		tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_FATAL,
			"failed to Get tbus head, dwGet:%u dwTail:%u dwSize:%u impossible!",
			pstVar->dwGet, pstVar->dwTail, pstVar->dwSize);
		pstVar->dwGet = pstVar->dwHead = pstVar->dwTail;
		return -1;
	}	


	/*检查数据长度与与队列中数据长度是否有效*/
	iPkgLen = pstHead->bHeadLen + pstHead->iBodyLen;
	if ((0 >= iPkgLen) || (iPkgLen > (int)pstVar->dwSize))
	{
		tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_ERROR,
			"invalid  length(%d) of pkg sending, its should not bigger than the size %d of channel queue or less than 1, impossible!",
			iPkgLen, pstVar->dwSize);
		pstVar->dwGet = pstVar->dwHead = pstVar->dwTail;
		return  -1;
	}
	if (iDataLen < iPkgLen) 
	{
		tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_FATAL,
			"tbusd_send_one_pkg, data len %d in TBUSCHANNEL is less than pkg len:%d, impossible!",
			iDataLen, iPkgLen);
		pstVar->dwGet = pstVar->dwHead = pstVar->dwTail;
		return -1;	
	}

	
	/*捎带接收数据包序列号*/
	pstPushVar = TBUS_CHANNEL_VAR_PUSH(a_pstChannel);	
	{
		LPTBUSHEAD pstTmpeHead = (LPTBUSHEAD)(sData +pstVar->dwGet);
		pstTmpeHead->bFlag |= TBUS_HEAD_FLAG_WITH_ACK;
		pstTmpeHead->iRecvSeq = htonl(pstPushVar->iRecvSeq);
	}
	

	/*发送数据*/
	a_pstConn->stSendMng.enRemainDataType = TBUSD_DATA_TYPE_BUSINESS;
	a_pstConn->stSendMng.iSendBuff = iPkgLen;
	a_pstConn->stSendMng.iSendPos = 0;
	a_pstConn->stSendMng.pszBussiData = sData +pstVar->dwGet;
	iRet = tbusd_send_remain_data(a_pstConn, (LPTBUSCHANNEL)a_pstChannel, a_pstRelayStat) ;
	if ( 0 != iRet )
	{
		LPRELAY pstRelay = a_pstConn->pstBindRelay;
		assert(NULL != pstRelay);
		tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_ERROR,
			"tbusd_send_remain_data failed to send remain data on socket %d of relay %d, ret %d",
			a_pstConn->iFd, pstRelay->iID, iRet ) ;		
	}

	return iRet;
}





int tbusd_proc_send(IN LPRELAYENV a_pstRunEnv, IN int a_iMaxSendPkg)
{
	int iRet = TBUS_SUCCESS;
	unsigned int i = 0;
	RELAY *ptRelay = NULL ;
	LPCONNNODE pstConn;
	int iSendPkg = 0;

	assert(NULL != a_pstRunEnv);

	for ( i=0; i< a_pstRunEnv->dwRelayCnt; i++ )
	{
		ptRelay = a_pstRunEnv->pastTbusRelay[i] ;
		if (!TBUSD_RELAY_IS_ENABLE(ptRelay))
		{
			continue ;
		}
		if ( RELAY_STATUS_ESTABLISHED != ptRelay->dwRelayStatus )
		{
			tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_TRACE, 
				"tbusd_proc_send, relay %d staus is %i should not send data", i, ptRelay->dwRelayStatus ) ;
			continue ;
		}

		iSendPkg += tbusd_send_pkg (a_pstRunEnv, ptRelay, a_iMaxSendPkg);
		pstConn = (LPCONNNODE)ptRelay->pConnNode;
		assert(NULL != pstConn);
		if (pstConn->bNeedFree)
		{
			tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_ERROR, 
				"tbusd_proc_send, relay<%d> send pkg failed %i, so close socket %d",
				ptRelay->iID, pstConn->iFd) ;
			tbusd_destroy_conn(a_pstRunEnv, pstConn);			
			continue ;
		}
	}/*for ( i=0; i< a_pstRunEnv->stRelayGlobal.dwRelayCnt; i++ )*/

	return iSendPkg ;
}

